package avicit.bdp.dds.server.worker.task.sqoop.generator.sources;

import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.datasource.DataSourceFactory;
import avicit.bdp.common.utils.DESUtils;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.dds.dispatch.enums.QueryType;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.bdp.dds.dispatch.task.sqoop.SqoopParameters;
import avicit.bdp.dds.dispatch.task.sqoop.sources.SourceMysqlParameter;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.server.entity.SqoopTaskExecutionContext;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.worker.task.sqoop.generator.ISourceGenerator;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * mysql source generator
 */
public class MysqlSourceGenerator implements ISourceGenerator {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public String generate(SqoopParameters sqoopParameters, TaskExecutionContext taskExecutionContext) {
        StringBuilder result = new StringBuilder();
        try {
            SourceMysqlParameter sourceMysqlParameter
                    = JSONUtils.parseObject(sqoopParameters.getSourceParams(), SourceMysqlParameter.class);

            SqoopTaskExecutionContext sqoopTaskExecutionContext = taskExecutionContext.getSqoopTaskExecutionContext();

            if (sourceMysqlParameter != null) {
                BaseDataSource baseDataSource = DataSourceFactory.getDatasource(sqoopTaskExecutionContext.getSourcetype(),
                        sqoopTaskExecutionContext.getSourceConnectionParams());
                if (baseDataSource != null) {
                    result.append(" --connect ")
                            .append(baseDataSource.getJdbcUrl())
                            .append(" --username ")
                            .append(baseDataSource.getUser())
                            .append(" --password ")
                            .append(DESUtils.decrypt(baseDataSource.getPassword()));

                    if (sourceMysqlParameter.getSrcQueryType() == QueryType.FORM.ordinal()) {
                        if (StringUtils.isNotEmpty(sourceMysqlParameter.getSrcTable())) {
                            result.append(" --table ").append(sourceMysqlParameter.getSrcTable());
                        }

                        if (StringUtils.isNotEmpty(sourceMysqlParameter.getSrcColumns())) {
                            result.append(" --columns ").append(sourceMysqlParameter.getSrcColumns());
                        }

                    } else if (sourceMysqlParameter.getSrcQueryType() == QueryType.SQL.ordinal()
                            && StringUtils.isNotEmpty(sourceMysqlParameter.getSrcQuerySql())) {
                        String srcQuery = sourceMysqlParameter.getSrcQuerySql();
                        if (srcQuery.toLowerCase().contains("where")) {
                            srcQuery += " AND " + "$CONDITIONS";
                        } else {
                            srcQuery += " WHERE $CONDITIONS";
                        }
                        result.append(" --query \'").append(srcQuery).append("\'");

                    }

                    List<Property> mapColumnHive = sourceMysqlParameter.getMapColumnHive();

                    if (mapColumnHive != null && !mapColumnHive.isEmpty()) {
                        StringBuilder columnMap = new StringBuilder();
                        for (Property item : mapColumnHive) {
                            columnMap.append(item.getProp()).append("=").append(item.getValue()).append(Constants.COMMA);
                        }

                        if (StringUtils.isNotEmpty(columnMap.toString())) {
                            result.append(" --map-column-param ")
                                    .append(columnMap.substring(0, columnMap.length() - 1));
                        }
                    }

                    List<Property> mapColumnJava = sourceMysqlParameter.getMapColumnJava();

                    if (mapColumnJava != null && !mapColumnJava.isEmpty()) {
                        StringBuilder columnMap = new StringBuilder();
                        for (Property item : mapColumnJava) {
                            columnMap.append(item.getProp())
                                    .append("=")
                                    .append(item.getValue())
                                    .append(Constants.COMMA);
                        }

                        if (StringUtils.isNotEmpty(columnMap.toString())) {
                            result.append(" --map-column-java ")
                                    .append(columnMap.substring(0, columnMap.length() - 1));
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage());
        }

        return result.toString();
    }
}
